/**
 * Copyright 2010 Sonal Goyal
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0 
 * 
 * Unless required by applicable law or agreed to in writing, software distributed 
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR 
 * CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and limitations under the License. 
 */

package com.meghsoft.hiho.avro;

import java.sql.Types;
import java.util.ArrayList;

import org.apache.avro.Schema;
import org.apache.log4j.Logger;

import com.meghsoft.hiho.mapreduce.lib.db.ColumnInfo;

public class DBMapper {
	
	private final static Logger logger = Logger
	.getLogger(com.meghsoft.hiho.avro.DBMapper.class);


	
	public static Schema getSchema(ArrayList<ColumnInfo> columnInfo) {
		ArrayList<Schema.Field> fields = new ArrayList<Schema.Field>();
		for (ColumnInfo column: columnInfo) {
			
			Schema.Field columnField = getColumnField(column);
			logger.debug("Schema for field " + column + " " + columnField.toString());
			fields.add(columnField);
		}
		Schema recordSchema = Schema.createRecord(fields);
		return recordSchema;
	}
	
	public static Schema.Field getColumnField(ColumnInfo column) {
		Schema fieldSchema = getSchema(column.getType());
		Schema.Field field = new Schema.Field(column.getName(), fieldSchema, null, null);
		return field;
	}
	
	// TODO: Verify mappings based on db definitions
	public static Schema getSchema(int type) {
		Schema.Type returnType = null;
		switch (type) {
			case Types.ARRAY: 
				returnType = Schema.Type.ARRAY;
		        break;
			case Types.BIGINT:
				returnType = Schema.Type.LONG;
		        break;
			case Types.BINARY:				
			case Types.BIT:
			case Types.BLOB:
				returnType = Schema.Type.BYTES;
				break;
			case Types.BOOLEAN:
				returnType = Schema.Type.BOOLEAN;
		        break;
			case Types.CHAR:
				returnType = Schema.Type.STRING;
		        break;
			case Types.CLOB:
				returnType = Schema.Type.BYTES;
		        break;
			//case Types.DATALINK
			case Types.DATE:
				returnType = Schema.Type.STRING;
		        break;
			case Types.DECIMAL:
				returnType = Schema.Type.DOUBLE;
		        break;
			//case Types.DISTINCT
			case Types.DOUBLE:
			case Types.FLOAT:
				returnType = Schema.Type.DOUBLE;
		        break;
			case Types.INTEGER:
				returnType = Schema.Type.INT;
		        break;
			//case Types.JAVA_OBJECT
			case Types.LONGVARBINARY:
				returnType = Schema.Type.BYTES;
		        break;
			case Types.LONGVARCHAR:
				returnType = Schema.Type.STRING;
		        break;
			case Types.NULL:
				returnType = Schema.Type.NULL;
		        break;
			case Types.NUMERIC:
				returnType = Schema.Type.DOUBLE;
		        break;
			//case Types.OTHER
			case Types.REAL:
				returnType = Schema.Type.FLOAT;
		        break;
			//case Types.REF
			case Types.SMALLINT:
				returnType = Schema.Type.INT;
		        break;
			//case Types.STRUCT
			case Types.TIME:
				returnType = Schema.Type.STRING;
		        break;
			case Types.TIMESTAMP:
				returnType = Schema.Type.STRING;
		        break;
			case Types.TINYINT:
				returnType = Schema.Type.INT;
		        break;
			case Types.VARBINARY:
				returnType = Schema.Type.BYTES;
		        break;
			case Types.VARCHAR :
				returnType = Schema.Type.STRING;
		        break;
			default:
				returnType = null;
				break;
		}
		Schema returnSchema = Schema.create(returnType);
		return returnSchema;
	}

}
